package cn.lsh.mapper;

import cn.lsh.writable.NetPage;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.StringTokenizer;

public class PageRankMapper extends Mapper<Text, Text, Text, NetPage> {
	/**当前页面*/
	private final NetPage currPage = new NetPage(0);
	/**关联网页*/
	private final NetPage relPage = new NetPage(1);
	private final Text mkey = new Text();


	@Override
	protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
		//输入数据样本有以下两种情况：
		//key：A			value:B D
		//key：A:0.5		value:B D
		//获取客户端设置的自定义参数
		final int runCount = context.getConfiguration().getInt("runCount", 1);

		if (runCount == 1) {
			currPage.setPage(key.toString());
		} else {
			StringTokenizer st = new StringTokenizer(key.toString(), ":");
			currPage.setPage(st.nextToken());
			currPage.setPr(Double.parseDouble(st.nextToken()));
		}
		currPage.setRelationPage(value.toString());
		mkey.set(currPage.getPage());
		//第一类数据，当前网站对应的网站关系，以及当前网站的pr值
		context.write(mkey, currPage);

		String[] users = currPage.getRelationPage();
		if (users != null && users.length > 0) {
			double avgPr = currPage.getPr() / users.length;
			for (int i = 0; i < users.length; i++) {
				relPage.setPage(users[i]);
				relPage.setPr(avgPr);
				mkey.set(users[i]);
				//第二类数据，当前网站对关联网站的投票，以及投票的权重
				context.write(mkey, relPage);
			}
		}

	}
}
